package com.taimanetworks.window;

import com.taimanetworks.kafka.parseTsp.Sourcedata;
import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

public class VhicleWindow extends RichWindowFunction<Sourcedata, List<Sourcedata>, Tuple, TimeWindow> {

/*    private ListState<Sourcedata> state = null;

    @Override
    public void open(Configuration parameters) throws Exception {

        state = getRuntimeContext().getListState(new ListStateDescriptor<Sourcedata>("my-list", Sourcedata.class));
    }*/

    //(KEY var1, W var2, Iterable<IN> var3, Collector<OUT> var4)
    @Override
    public void apply(Tuple tuple, TimeWindow window, Iterable<Sourcedata> input, Collector<List<Sourcedata>> out) throws Exception {
        List<Sourcedata> list = IteratorUtils.toList(input.iterator());
        if(list.size()>0){
            out.collect(list);
        }

    }


}
